package com.atguigu.app

import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis
import org.json4s.native.Serialization

import collection.JavaConverters._
object SaleDetailApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SaleDetailApp")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf,Seconds(5))

    //3.消费kafka中的数据分别获取订单表的数据以及订单明细表的数据
    val orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER,ssc)

    val orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL,ssc)

    //4.将数据转为样例类,并返回k，v类型的数据，key是两条流的关联条件 即orderId
    val orderInfoDStream = orderInfoKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])

        //补全字段
        orderInfo.create_date = orderInfo.create_time.split(" ")(0)
        orderInfo.create_hour = orderInfo.create_time.split(" ")(1).split(":")(0)
        //对手机号做脱敏操作
        (orderInfo.id,orderInfo)
      })
    })

    val orderDetailDStream= orderDetailKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        (orderDetail.order_id,orderDetail)
      })
    })
//    orderInfoDStream.print()
//    orderDetailDStream.print()

    //5.通过join，对两条流进行合并
//    val joinDStream: DStream[(String, (OrderInfo, OrderDetail))] = orderInfoDStream.join(orderDetailDStream)
    val fullJoinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDStream.fullOuterJoin(orderDetailDStream)

    //6.采用加缓存的方式处理因网络延迟所带来的数据丢失问题
    val noUserSaleDetailDStream: DStream[SaleDetail] = fullJoinDStream.mapPartitions(partition => {
      implicit val formats = org.json4s.DefaultFormats

      //创建List集合用来存放关联后的结果数据（SaleDetail）
      val details: util.ArrayList[SaleDetail] = new util.ArrayList[SaleDetail]()

      //创建redis连接
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      partition.foreach { case (orderId, (infoOpt, detailOpt)) =>

        //存放orderinfo数据的rediskey
        val orderInfoRedisKey: String = "orderInfo:" + orderId
        //存放orderDetail数据的RedisKey
        val orderDetailRedisKey: String = "orderDetail:" + orderId
        //a.判断orderInfo是否存在
        if (infoOpt.isDefined) {
          //orderInfo表存在
          //a.2取出订单表数据
          val orderInfo: OrderInfo = infoOpt.get

          //a.3判断orderDetail数据是否存在
          if (detailOpt.isDefined) {
            //orderDetail表存在
            //a.4取出orderDetail数据
            val orderDetail: OrderDetail = detailOpt.get
            //a.5关联两个表的数据，将其写入SaleDetail样例类中
            val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
            //a.6将关联后的SaleDetail存入结果集合中
            details.add(detail)
          }

          //b.1无论如何都要将orderInfo数据写入redis缓存
          val orderInfoJson: String = Serialization.write(orderInfo)

          //          val orderInfoJsonStr: String = JSON.toJSONString(orderInfo)
          jedis.set(orderInfoRedisKey, orderInfoJson)

          //b.2给存入redis中的orderinfo数据设置过期时间，通常过期时间是网络延迟时间（建议：比这个时间多几秒）
          jedis.expire(orderInfoRedisKey, 20)

          //c.1查询对方缓存中是否有能够关联上的数据
          if (jedis.exists(orderDetailRedisKey)) {
            //证明有能够关联上的OrderDetail数据
            //c.2获取redis缓存中能够关联上的orderDetail数据
            val detailSet: util.Set[String] = jedis.smembers(orderDetailRedisKey)
            //c.3遍历set集合获取到每一个OrderDetail数据
            for (elem <- detailSet.asScala) {
              //将查询出来的Json字符串转为样例类
              val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              //c.4将关联后的样例类写入结果集合
              details.add(detail)
            }
          }
        } else {
          /**
            * if 存在 ->获取到orderDetail数据->查询orderinfo缓存中是否有能关联上的数据 ->能关联上则写入结果集合
            * | else
            * v
            * 没有能关联上，则将自己写入缓存
            */
          //orderinfo不在
          //d.1判断orderDetail数据是否存在
          if (detailOpt.isDefined) {
            //OrderDetail数据存在
            //取出orderdetail数据
            val orderDetail: OrderDetail = detailOpt.get
            //e.查询orderInfo缓存中是否有能关联上的数据
            if (jedis.exists(orderInfoRedisKey)) {
              //有能关联上的orderInfo数据
              //e.2取出orderInfo数据
              val infoJsonStr: String = jedis.get(orderInfoRedisKey)
              //e.3将读出来的json字符串转为样例类
              val orderInfo: OrderInfo = JSON.parseObject(infoJsonStr, classOf[OrderInfo])
              //关联数据写入SaleDetail样例类
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              details.add(detail)
            } else {
              //没有能关联上的orderInfo数据
              //f.将自己写入缓存
              //f.2将OrderDetail样例类转为Json字符串
              val orderDetailJsonStr: String = Serialization.write(orderDetail)
              jedis.sadd(orderDetailRedisKey, orderDetailJsonStr)
              //对orderDetail数据设置过期时间，以防redis内存溢出
              jedis.expire(orderDetailRedisKey, 20)
            }
          }
        }
      }
      jedis.close()
      //将集合中的数据返回，同时将集合转为迭代器
      details.asScala.toIterator
    })

    //7.反查userInfo缓存关联用户表数据
    val saleDetailDStream: DStream[SaleDetail] = noUserSaleDetailDStream.mapPartitions(partition => {
      //创建redis连接
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      val details: Iterator[SaleDetail] = partition.map(saleDetail => {
        //查询userInfo的数据
        val userInfoRedisKey: String = "userInfo:" + saleDetail.user_id
        val userInfoJsonStr: String = jedis.get(userInfoRedisKey)
        //将查询到的json字符串转为样例类
        val userInfo: UserInfo = JSON.parseObject(userInfoJsonStr, classOf[UserInfo])
        //关联用户表的数据
        saleDetail.mergeUserInfo(userInfo)
        saleDetail
      })
      jedis.close()
      details
    })

    saleDetailDStream.print()
    //8.将数据写入ES
    saleDetailDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, SaleDetail)] = partition.toList.map(saleDetail => {
          (saleDetail.order_detail_id, saleDetail)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_INDEX_SALEDETAIL+"211027",list)
      })
    })

    ssc.start()
    ssc.awaitTermination()

  }

}
